"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : SQL_DDL_Drop_Owned
Case Name   : DROP OWNED删除的对象类型验证
Create At   : 2022/08/24
Owner       : opentestcase019
Description :
    1.创建新用户
    2.新用户创建表
    3.新用户创建视图
    4.新用户创建索引
    5.新用户创建序列
    6.新用户创建目录对象
    7.新用户创建函数
    8.新用户创建触发器
    9.新用户创建存储过程
    10.新用户创建模式
    11.删除新用户所拥有的对象
    12.删除用户，恢复参数
Expect      :
    1.创建成功
    2.创建成功
    3.创建成功
    4.创建成功
    5.创建成功
    6.创建成功
    7.创建成功
    8.创建成功
    9.创建成功
    10.创建成功
    11.删除成功
    12.删除成功，恢复成功
History     :
    Modified by opentestcase016 2022/11/17:优化用例，保证\d查询结果匹配
    Modified by opentestcase016 2023/2/3:优化用例，保证断言匹配正确
"""
import os
import unittest
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Common import Common
from yat.test import Node
from yat.test import macro


class DropOwned(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.user_node = Node('PrimaryDbUser')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.user_name = 'u_drop_owned_0010'
        self.table_name = 't_drop_owned_0010'
        self.view_name = 'v_drop_owned_0010'
        self.index_name = 'i_drop_owned_0010'
        self.seq_name = 'seq_drop_owned_0010'
        self.dir_object_name = 'dir_object_drop_owned_0010'
        self.func_name = 'func_drop_owned_0010'
        self.trigger_name = 'trigger_drop_owned_0010'
        self.procedure_name = 'procedure_drop_owned_0010'
        self.schema_name = 'schema_drop_owned_0010'
        self.config_directory = 'enable_access_server_directory'
        self.directory_default = self.common.show_param(self.config_directory)

    def test_drop_owned(self):
        text = "-----step1:创建用户;expect:创建成功-----"
        self.logger.info(text)
        create_user = f"drop user if exists {self.user_name};" \
            f"create user {self.user_name} " \
            f"sysadmin password '{macro.COMMON_PASSWD}';"
        create_msg = self.primary_sh.execut_db_sql(create_user)
        self.logger.info(create_msg)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)

        text = "-----step2:新用户创建表;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"drop table if exists {self.table_name};" \
            f"create table {self.table_name}(id int, name varchar(15));" \
            f"insert into {self.table_name} " \
            f"values(1, '软件开发'),(2, '软件测试');"
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_msg = self.primary_sh.execut_db_sql(
            '\\d', f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.table_name, sql_msg.splitlines()[-2], '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step3:新用户创建视图;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"drop view if exists {self.view_name};" \
            f"create view {self.view_name} as " \
            f"select * from {self.table_name};"
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_VIEW_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.CREATE_VIEW_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_msg = self.primary_sh.execut_db_sql(
            '\\d', f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.view_name, sql_msg.splitlines()[-2], '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step4:新用户创建索引;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"drop index if exists {self.index_name};" \
            f"create unique index {self.index_name} on {self.table_name}(id);"
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_INDEX_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.CREATE_INDEX_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_msg = self.primary_sh.execut_db_sql(
            '\\di', f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.index_name, sql_msg.splitlines()[-2], '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step5:新用户创建序列;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"drop sequence if exists {self.seq_name};" \
            f"create sequence {self.seq_name} start 101 cache 20;"
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_SEQUENCE_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertIn(self.constant.CREATE_SEQUENCE_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_msg = self.primary_sh.execut_db_sql(
            '\\d', f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertGreater(len(sql_msg.splitlines()), 4, '执行失败' + text)
        self.assertIn(self.seq_name, sql_msg.splitlines()[-4], '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-4], '执行失败' + text)

        text = "-----step6:新用户创建目录对象;expect:创建成功-----"
        self.logger.info(text)
        if 'on' != self.directory_default:
            result = self.primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_directory}=on')
            self.assertTrue(result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_directory)
        self.assertEquals(show_msg, 'on', '执行失败' + text)
        sql_cmd = f'''create or replace directory {self.dir_object_name} 
            as '/tmp';
            select t1.dirname, t2.usename, t1.dirpath from pg_directory t1 
            inner join pg_user t2 on t2.usesysid=t1.owner
            where dirname='{self.dir_object_name}';'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.CREATE_DIRECTORY_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.dir_object_name, sql_msg.splitlines()[-2],
                      '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step7:新用户创建函数;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f'''create or replace function {self.func_name}() 
        returns trigger as 
        \$\$ 
        declare 
        begin 
            update {self.table_name} set name='开发工程师' where id=2; 
        return new;
        end 
        \$\$ language plpgsql;
        select t1.proname, t2.usename from pg_proc t1 inner join pg_user t2 
        on t2.usesysid=t1.proowner where proname='{self.func_name}';'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.func_name, sql_msg.splitlines()[-2], '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step8:新用户创建触发器;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f'''drop trigger if exists {self.trigger_name}
            on {self.table_name};'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.TRIGGER_DROP_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_cmd = f'''create trigger {self.trigger_name}
            after insert on {self.table_name} for each row 
            execute procedure {self.func_name}();
            select t1.tgname, t2.usename  from pg_trigger t1 
            inner join pg_user t2 on t2.usesysid=t1.tgowner 
            where tgname='{self.trigger_name}';'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.TRIGGER_CREATE_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.trigger_name, sql_msg.splitlines()[-2],
                      '执行失败' + text)
        self.assertIn(self.user_name,  sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step9:新用户创建存储过程;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f'''create or replace procedure 
        {self.procedure_name}(tb_name varchar) is 
        begin 
            execute immediate 'insert into  '|| tb_name || ' values(3,''西安'');';
        end;'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.CREATE_PROCEDURE_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_cmd = f'''select t1.proname, t2.usename 
            from pg_proc t1 inner join pg_user t2 on t2.usesysid=t1.proowner 
            where proname='{self.procedure_name}';'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.procedure_name, sql_msg.splitlines()[-2],
                      '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step10:新用户创建模式;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f'''drop schema if exists {self.schema_name};'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        sql_cmd = f'''create schema {self.schema_name};
            select t1.nspname, t2.usename from pg_namespace t1 
            inner join pg_user t2 on t2.usesysid=t1.nspowner 
            where t1.nspname='{self.schema_name}';'''
        sql_msg = self.primary_sh.execut_db_sql(
            sql_cmd, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG,
                      sql_msg, '执行失败' + text)
        self.assertGreater(len(sql_msg.splitlines()), 2, '执行失败' + text)
        self.assertIn(self.schema_name, sql_msg.splitlines()[-2],
                      '执行失败' + text)
        self.assertIn(self.user_name, sql_msg.splitlines()[-2], '执行失败' + text)

        text = "-----step11:删除新用户所拥有的对象，查询是否删除成功;expect:删除成功-----"
        self.logger.info(text)
        sql_cmd_1 = f"drop owned by {self.user_name} cascade;\\d"
        sql_msg_1 = self.primary_sh.execut_db_sql(
            sql_cmd_1, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg_1)
        self.assertIn(self.constant.DROP_OWNED_SUCCESS, sql_msg_1,
                      '执行失败' + text)
        self.assertIn('No relations found', sql_msg_1, '执行失败' + text)
        sql_cmd_2 = f'''select dirname, owner, dirpath from pg_directory t1 
            inner join pg_user t2 on t2.usesysid=t1.owner 
            where dirname='{self.dir_object_name}';
            select t1.proname, t2.usename from pg_proc t1 inner join pg_user t2 
            on t2.usesysid=t1.proowner where proname='{self.func_name}';
            select t1.tgname, t2.usename  from pg_trigger t1 
            inner join pg_user t2 on t2.usesysid=t1.tgowner 
            where tgname='{self.trigger_name}';
            select t1.proname, t2.usename from pg_proc t1 inner join pg_user t2 
            on t2.usesysid=t1.proowner where proname='{self.procedure_name}';
            select t1.nspname, t2.usename from pg_namespace t1 
            inner join pg_user t2 on t2.usesysid=t1.nspowner 
            where t1.nspname='{self.schema_name}';
            \\di'''
        sql_msg_2 = self.primary_sh.execut_db_sql(
            sql_cmd_2, f'-U {self.user_name} -W \'{macro.COMMON_PASSWD}\'')
        self.logger.info(sql_msg_2)
        self.assertEquals(5, sql_msg_2.count('(0 rows)'), '执行失败' + text)
        self.assertIn('No relations found', sql_msg_2, '执行失败' + text)

    def tearDown(self):
        text = "-----step12:删除用户，恢复参数;expect:删除成功，恢复成功-----"
        self.logger.info(text)
        drop_cmd = f"drop user {self.user_name} cascade;"
        drop_msg = self.primary_sh.execut_db_sql(drop_cmd)
        self.logger.info(drop_msg)
        current = self.common.show_param(self.config_directory)
        if self.directory_default != current:
            self.primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_directory}={self.directory_default}')
        show_msg = self.common.show_param(self.config_directory)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text)
        self.assertEquals(show_msg, self.directory_default, '执行失败' + text)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
